package com.ouwen.smartpay.service.chain.slots.limiting;

import com.fm.common.emuns.ResponseCode;
import com.fm.core.common.boot.Apps;
import com.fm.core.common.exception.LimitingCheckException;
import com.fm.core.common.facade.BizResultBase;
import com.fm.core.utils.Parks;
import com.fm.core.utils.Strings;
import com.fm.safety.limiting.FailStrategy;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.RateLimiter;
import com.ouwen.smartpay.openapi.order.ServiceOrder;
import com.ouwen.smartpay.service.chain.context.Context;
import com.ouwen.smartpay.service.chain.slotchain.AbstractLinkedProcessorSlot;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

/**
 * @author <a href="http://youngitman.tech">青年IT男</a>
 * @version v1.0.0
 * @className LimitingSlot
 * @description 服务限流
 * @JunitTest: {@link  }
 * @date 2020-04-26 10:14
 **/
public class LimitingSlot extends AbstractLinkedProcessorSlot<ServiceOrder> {

    /*** 资源和限流器关系 */
    public static final Map<String, RateLimiter> resouceRateLimiterMap = Maps.newConcurrentMap();

    /*** 每秒通过量 */
    private int secLimit;

    /*** 超时时间 单位/描述 */
    private int timeout;

    private FailStrategy failStrategy;

    @Override
    public void entry(Context context, ServiceOrder order, Object... args) throws Throwable {

        Logger logger = LoggerFactory.getLogger(getClass());

        try {

            //每秒通过量
            secLimit = Integer.valueOf(Apps.getEnvironment().getProperty("fm.limit.sec.count", "100"));

            //阻塞等待时间
            timeout = Integer.valueOf(Apps.getEnvironment().getProperty("fm.limit.timeout", "60"));

            //请求失败处理策略
            failStrategy = FailStrategy.findStatus((Apps.getEnvironment().getProperty("fm.limit.fail.strategy", "FASTFAIL")));

            //执行限流规则
            doLimiting(secLimit, failStrategy, timeout, context);

            fireEntry(context, order, args);

        } catch (Exception e) {

            logger.warn(String.format("商户号【%s】子商户号【%s】调用的服务【%s】太频繁,请稍后重试!:", context.getPartnerId(), context.getSubPartnerId(), context.getService()));

            BizResultBase result = new BizResultBase();

            result.setRespCode(ResponseCode.FAIL.code());

            result.setRespMsg(ResponseCode.FAIL.message());

            result.setDescription(e.getMessage());

            exit(context, order, args, result);
        }

    }

    @Override
    public void exit(Context context, Object... args) {

        fireExit(context, args);

    }

    /***
     *
     * 构建限流key 目前限流基本：商户号+子商户号+service
     *
     * @author liyong
     * @date 9:46 AM 2020/5/7
     * @param context
     * @exception
     * @return java.lang.String
     **/
    private String buildRelateKey(Context context) {

        StringBuilder sb = new StringBuilder();

        sb.append(context.getPartnerId()).append(":");

        if (Strings.isNotBlank(context.getSubPartnerId())) {

            sb.append(context.getSubPartnerId()).append(":");

        }

        sb.append(context.getService());

        return sb.toString();
    }

    /***
     *
     * 根据限流key、失败策略和超时时间进行限流操作
     *
     * @author liyong
     * @date 9:54 AM 2020/5/7
     * @param limiting
     * @param failStrategy
     * @param timeout
     * @param context
     * @exception
     * @return void
     **/
    private void doLimiting(int limiting, FailStrategy failStrategy, long timeout, Context context) {

        //构建限流key
        String key = buildRelateKey(context);

        resouceRateLimiterMap.computeIfAbsent(key, (k) -> RateLimiter.create(limiting, 1000, TimeUnit.MILLISECONDS));

        RateLimiter rateLimiter = resouceRateLimiterMap.get(key);

        boolean acquired;

        if (FailStrategy.BLOCKING == failStrategy) {

            //最大阻塞120秒，需要根据dubbo调用最大超时时间设定
            if (timeout > 0 && timeout < 120) {

                //自旋等待超时
                acquired = timeCheck(timeout, () -> rateLimiter.tryAcquire());

            } else {

                //最大阻塞60秒,RPC调用超时时间相关
                rateLimiter.tryAcquire(60, TimeUnit.SECONDS);

                acquired = true;
            }

        } else {

            acquired = rateLimiter.tryAcquire();

        }
        if (!acquired) {

            throw new LimitingCheckException(new Throwable(String.format("商户号【%s】子商户号【%s】调用的服务【%s】太频繁,请稍后重试!:", context.getPartnerId(), context.getSubPartnerId(), context.getService())));

        }

    }


    /***
     *
     * 超时检测（自旋）
     *
     * @author liyong
     * @date 11:44 2020-04-16
     * @param timeout 单位秒
     * @param supplier
     * @exception
     * @return boolean
     **/
    private boolean timeCheck(long timeout, Supplier<Boolean> supplier) {

        boolean acquired = false;

        if (timeout > 0) {

            //纳秒
            long timout = timeout * 1000 * 1000 * 1000;

            long start = System.nanoTime();

            while (true) {

                long current = System.nanoTime();

                if (current - start < timout) {

                    //100毫秒
                    Parks.park(100, TimeUnit.MILLISECONDS);

                    acquired = supplier.get();

                    if (acquired) {

                        break;

                    }

                } else {

                    acquired = false;

                    break;
                }
            }
        }
        return acquired;
    }
}
